package com.atguigu.day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

public class Example11 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 9999)
                .keyBy(r -> 1)
                .process(new TimerProcess())
                .print();

        env.execute();
    }

    public static class TimerProcess extends KeyedProcessFunction<Integer, String, String> {
        @Override
        public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
            // value到达process算子的机器时间
            long currTs = ctx.timerService().currentProcessingTime();
            // 10秒钟之后的时间戳
            long tenSecondLater = currTs + 30 * 1000L;
            // 注册一个时间戳是tenSecondLater的定时器
            ctx.timerService().registerProcessingTimeTimer(tenSecondLater);
            out.collect("数据：" + value + " 到达，到达时间是：" + new Timestamp(currTs) + "，" +
                    "注册了一个时间戳是：" + new Timestamp(tenSecondLater) + " 的定时器");
        }

        // 当机器时间到达tenSecondLater时，触发onTimer的调用
        // timestamp参数就是上面的tenSecondLater
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            out.collect("定时器触发了！触发时间是：" + new Timestamp(timestamp));
        }
    }
}
